package com.flinksql.test;

import com.flinksql.bean.WaterSensor;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author: Lin
 * @create: 2021-07-07 16:12
 * @description:
 **/
public class FlinkSQL_Test8_OverWindow {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("hadoop102", 9999);
        SingleOutputStreamOperator<WaterSensor> mapDS = source.map(new RichMapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0]
                        , Long.parseLong(split[1])
                        , Integer.parseInt(split[2]));
            }
        });

        //1.创建表的执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //2.创建表：将流转换成动态表。
        Table table = tableEnv.fromDataStream(mapDS,$("id"),$("ts"),$("vc"),$("pt").proctime());

        //3.把表注册为一个临时视图
        tableEnv.createTemporaryView("sensor",table);

        //ROW/RANGE not allowed with RANK, DENSE_RANK or ROW_NUMBER functions
        tableEnv.sqlQuery("select * from (SELECT *," +
                " ROW_NUMBER() OVER (PARTITION BY id ORDER BY pt desc) AS row_num" +  //全窗口
                " FROM sensor) where row_num=1")
                .execute()
                .print();
    }
}
